package com.example.kafka.controller;

import java.util.UUID;

import org.springframework.http.ResponseEntity;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.example.kafka.service.IdempotentMessageService;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@RestController
@RequestMapping("/api/idempotent")
@RequiredArgsConstructor
public class IdempotentController {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final IdempotentMessageService idempotentMessageService;

    /**
     * 测试用例1：订单创建（基于数据库唯一索引的幂等处理）
     */
    @PostMapping("/order")
    public ResponseEntity<String> createOrder() {
        
        String orderNo = "ORDER-" + UUID.randomUUID().toString();
        kafkaTemplate.send("order-create", orderNo);
        return ResponseEntity.ok("订单创建消息已发送: " + orderNo);
    }

    /**
     * 测试用例2：积分发放（基于Redis的幂等处理）
     */
    @PostMapping("/points")
    public ResponseEntity<String> awardPoints() {
        String message = "POINTS-" + UUID.randomUUID().toString();
        kafkaTemplate.send("points-award", message);
        return ResponseEntity.ok("积分发放消息已发送: " + message);
    }

    /**
     * 测试用例3：支付状态更新（基于业务状态的幂等处理）
     */
    @PostMapping("/payment/{orderNo}")
    public ResponseEntity<String> updatePaymentStatus(@PathVariable String orderNo) {
        kafkaTemplate.send("payment-status", orderNo);
        return ResponseEntity.ok("支付状态更新消息已发送: " + orderNo);
    }

    /**
     * 测试用例4：通知发送（基于消息ID的幂等处理）
     */
    @PostMapping("/notification")
    public ResponseEntity<String> sendNotification() {
        String messageId = "NOTIFY-" + UUID.randomUUID().toString();
        kafkaTemplate.send("notification", messageId);
        return ResponseEntity.ok("通知发送消息已发送: " + messageId);
    }

    /**
     * 测试用例5：库存扣减（基于分布式锁的幂等处理）
     */
    @PostMapping("/inventory/{productId}")
    public ResponseEntity<String> deductInventory(@PathVariable String productId) {
        kafkaTemplate.send("inventory-deduct", productId);
        return ResponseEntity.ok("库存扣减消息已发送: " + productId);
    }

    /**
     * 模拟重复发送消息（用于测试幂等性）
     */
    @PostMapping("/repeat/{type}/{id}")
    public ResponseEntity<String> repeatMessage(
            @PathVariable String type,
            @PathVariable String id) {
        String topic;
        String message;

        switch (type.toLowerCase()) {
            case "order":
                topic = "order-create";
                message = id;
                break;
            case "points":
                topic = "points-award";
                message = id;
                break;
            case "payment":
                topic = "payment-status";
                message = id;
                break;
            case "notification":
                topic = "notification";
                message = id;
                break;
            case "inventory":
                topic = "inventory-deduct";
                message = id;
                break;
            default:
                return ResponseEntity.badRequest().body("不支持的消息类型: " + type);
        }

        kafkaTemplate.send(topic, message);
        return ResponseEntity.ok("重复消息已发送: " + message);
    }
}
